package cn.itcast;

import cn.itcast.es.pojo.Goods;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.Avg;
import org.elasticsearch.search.aggregations.metrics.InternalAvg;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.*;

public class ElasticDemo {

    private RestHighLevelClient client;

    /**
     * 建立连接
     */
    @Before
    public void init() throws IOException {
        client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("192.168.206.99", 9200, "http")
                )
        );
    }

    /**
     * 关闭客户端连接
     */
    @After
    public void close() throws IOException {
        client.close();
    }

    @Test
    public void testCreateIndex() throws IOException {

        CreateIndexRequest createIndexRequest = new CreateIndexRequest("goods");

        createIndexRequest.source("{\n" +
                "  \"settings\": {\n" +
                "    \"analysis\": {\n" +
                "      \"analyzer\": {\n" +
                "        \"my_pinyin\": {\n" +
                "          \"tokenizer\": \"ik_smart\",\n" +
                "          \"filter\": [\n" +
                "            \"py\"\n" +
                "          ]\n" +
                "        }\n" +
                "      },\n" +
                "      \"filter\": {\n" +
                "        \"py\": {\n" +
                "          \"type\": \"pinyin\",\n" +
                "          \"keep_full_pinyin\": false,\n" +
                "          \"keep_joined_full_pinyin\": true,\n" +
                "          \"keep_original\": true,\n" +
                "          \"limit_first_letter_length\": 16,\n" +
                "          \"remove_duplicated_term\": true\n" +
                "        }\n" +
                "      }\n" +
                "    }\n" +
                "  },\n" +
                "  \"mappings\": {\n" +
                "    \"properties\": {\n" +
                "      \"id\": {\n" +
                "        \"type\": \"keyword\"\n" +
                "      },\n" +
                "      \"name\": {\n" +
                "        \"type\": \"completion\",\n" +
                "        \"analyzer\": \"my_pinyin\",\n" +
                "        \"search_analyzer\": \"ik_smart\"\n" +
                "      },\n" +
                "      \"title\":{\n" +
                "        \"type\": \"text\",\n" +
                "        \"analyzer\": \"my_pinyin\",\n" +
                "        \"search_analyzer\": \"ik_smart\"\n" +
                "      },\n" +
                "      \"price\":{\n" +
                "        \"type\": \"long\"\n" +
                "      }\n" +
                "    }\n" +
                "  }\n" +
                "}", XContentType.JSON);

        CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);

        boolean acknowledged = createIndexResponse.isAcknowledged();

        System.out.println("acknowledged = " + acknowledged);
    }

    @Test
    public void testAddDocuments() throws IOException {


        // 1.准备文档数据
        List<Goods> list = new ArrayList<>();
        list.add(new Goods(1L, "红米9", "红米9手机 数码", 1499L));
        list.add(new Goods(2L, "三星 Galaxy A90", "三星 Galaxy A90 手机 数码 疾速5G 骁龙855", 3099L));
        list.add(new Goods(3L, "Sony WH-1000XM3", "Sony WH-1000XM3 降噪耳机 数码", 2299L));
        list.add(new Goods(4L, "松下剃须刀", "松下电动剃须刀高转速磁悬浮马达", 599L));
        // 2.创建BulkRequest对象
        BulkRequest bulkRequest = new BulkRequest();
        // 3.创建多个IndexRequest对象，并添加到BulkRequest中
        for (Goods goods : list) {
            bulkRequest.add(new IndexRequest("goods")
                    //id的类型String，lucene，中数值存储采用转意存储，132====》口@口,转意后的查询会带来问题，"132"
                    .id(goods.getId().toString())
                    .source(JSON.toJSONString(goods), XContentType.JSON)
            );
        }
        // 4.发起请求
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);

        System.out.println("status: " + bulkResponse.status());

    }


    @Test
    public void testQuery() throws IOException {

        SearchRequest searchRequest = new SearchRequest("goods");

        //查询，分页，排序，高亮，聚合   武大
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        //添加查询条件
        sourceBuilder.query(QueryBuilders.matchQuery("title", "数码").operator(Operator.AND));

        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.preTags("<em>");
        highlightBuilder.postTags("</em>");
        highlightBuilder.field("title");

        //添加高亮条件
        sourceBuilder.highlighter(highlightBuilder);

        //添加分页条件
        sourceBuilder.from(0);
        sourceBuilder.size(2);

        //添加排序条件
        sourceBuilder.sort(SortBuilders.fieldSort("price").order(SortOrder.DESC));
        sourceBuilder.sort(SortBuilders.fieldSort("_score").order(SortOrder.DESC));


        //添加聚合条件
        sourceBuilder.aggregation(AggregationBuilders.terms("price_agg").field("price")
                .subAggregation(AggregationBuilders.avg("avg_price").field("price")));

        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);


        //搜索结果
        SearchHits hits = searchResponse.getHits();

        //命中的数量
        long totalHits = hits.getTotalHits().value;

        //查询结果
        SearchHit[] searchHits = hits.getHits();

        for (SearchHit searchHit : searchHits) {

            //取出高亮结果
            HighlightField highlightField = searchHit.getHighlightFields().get("title");

            //把高亮结果封装为字符串
            String hightTitle = StringUtils.join(highlightField.getFragments());

            Goods goods = JSON.parseObject(searchHit.getSourceAsString(), Goods.class);

            goods.setTitle(hightTitle);


            float score = searchHit.getScore();

            System.out.println("得分:" + score + " 内容：" + goods);
        }


        //聚合结果
        Aggregations aggregations = searchResponse.getAggregations();

        //聚合分桶的结果都可以交由terms来处理
        Terms aggregation = aggregations.get("price_agg");

        aggregation.getBuckets().forEach(bucket -> {
            long key = bucket.getKeyAsNumber().longValue();
            long docCount = bucket.getDocCount();


            Avg avg_price = bucket.getAggregations().get("avg_price");

            double value = avg_price.getValue();

            System.out.println(key + "有" + docCount + " 个平均价格" + value);
        });
    }


    @Test
    public void testSuggest() throws IOException {

        Set<String> result = new HashSet<>();

        SearchRequest searchRequest = new SearchRequest("articles");

        //武大
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        //补全联想条件构造器
        SuggestBuilder suggestBuilder = new SuggestBuilder();

        //添加补全条件
        suggestBuilder.addSuggestion("hehe", SuggestBuilders.completionSuggestion("heima116").prefix("e"));

        //把联想条件加入，武大
        sourceBuilder.suggest(suggestBuilder);

        //把武大封装的 所有 的条件，加入到查询条件对象中
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);

        Suggest suggest = searchResponse.getSuggest();
        Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>> suggestion = suggest.getSuggestion("hehe");

        suggestion.forEach(sugg -> {
            sugg.getOptions().forEach(opt -> {
                result.add(opt.getText().toString());
            });
        });

        System.out.println(result);
    }


    @Test
    public void testMonoSubScribe() {

        testGetDocumentByIdAsync().subscribe(System.out::print);

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    public Mono<Goods> testGetDocumentByIdAsync() {

        return Mono.create(sink -> {
            System.out.println("准备开始查询");
            // 准备一个查询文档的请求
            GetRequest request = new GetRequest("goods", "1");
            // 异步查询一个文档，耗时50ms
            client.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                @Override
                public void onResponse(GetResponse response) {
                    // 获取source
                    String json = response.getSourceAsString();
                    // 把json反序列化
                    Goods goods = JSON.parseObject(json, Goods.class);

                    sink.success(goods);
                    System.out.println("查询结束，得到结果： " + goods);
                }

                @Override
                public void onFailure(Exception e) {
                    sink.error(e);
                }
            });
        });
    }

}